#include "PacketWriter.h"

#include <packet/Packet.h>
#include <PacketListener.h>
#include <PacketFilter.h>
#include <PacketInterceptor.h>
#include <XMPPConnection.h>

//#include <util/Thread.h>


PacketWriter::PacketWriter(XMPPConnection * connection) : 
	connection(connection),
	writer(connection->getWriter()) {
    done = false;
}

void PacketWriter::init() {
    //writer = connection->writer;
    
}

void PacketWriter::sendPacket(Packet * packet) {
    if (!done) {
    	// Invoke interceptors for the new packet that is about to be sent. Interceptors
        // may modify the content of the packet.
        processInterceptors(packet);
        
		queue.put(packet);
        queue.notifyAll();

        // Process packet writer listeners. Note that we're using the sending
        // thread so it's expected that listeners are fast.
        processListeners(packet);
    }
}

void PacketWriter::addPacketListener(PacketListener * packetListener, PacketFilter * packetFilter) {
    listeners[packetListener] = new ListenerWrapper(packetListener, packetFilter);
}

void PacketWriter::removePacketListener(PacketListener * packetListener) {
	std::map<PacketListener*, ListenerWrapper*>::iterator it = listeners.find(packetListener);
	ListenerWrapper * wrapper = it->second();
	listeners.remove(packetListener);
	delete packetListener;
	delete wrapper;
}

int PacketWriter::getPacketListenerCount() {
	return listeners.size();
}

void PacketWriter::addPacketInterceptor(PacketInterceptor * packetInterceptor, PacketFilter * packetFilter) {
	interceptors[packetInterceptor] = new InterceptorWrapper(packetInterceptor, packetFilter);
}

void PacketWriter::removePacketInterceptor(PacketInterceptor * packetInterceptor) {
	InterceptorWrapper * wrapper = interceptors[packetInterceptor];
	interceptors.erase(packetInterceptor);
	delete packetInterceptor;
	delete wrapper;
}

void PacketWriter::startup() {
	boost::thread writerThread(boost::thread(boost::bind(&PacketWriter::writePackets, this)));
	writerThread.join();
}

void PacketWriter::startKeepAliveProcess() {
    // Schedule a keep-alive task to run if the feature is enabled. will write
    // out a space character each time it runs to keep the TCP/IP connection open.
    int keepAliveInterval = SmackConfiguration::getKeepAliveInterval();
    if (keepAliveInterval > 0) {
        Thread * keepAliveThread = new KeepAliveTask(keepAliveInterval);
        keepAliveThread.setDaemon(true);
        keepAliveThread.setName("Smack Keep Alive (" + connection.connectionCounterValue + ")");
        keepAliveThread.start();
    }
}

void PacketWriter::setStream(std::ostream * dataStream) {
	writer = dataStream;
}

void PacketWriter::shutdown() {
    done = true;
    queue.notifyAll();
}

void PacketWriter::cleanup() {
	interceptors.clear();
	listeners.clear();
}

PacketWriter::WriterThread::WriterThread(PacketWriter * packetWriter) :
	packetWriter(packetWriter)
{
}

void PacketWriter::WriterThread::run(void * arg) {
	packetWriter->writePackets(this);
}

PacketWriter::KeepAliveTask::KeepAliveTask(PacketWriter * packetWriter, int delay) :
	packetWriter(packetWriter),
	delay(delay)
{
}

void PacketWriter::KeepAliveTask::run(void * arg) {
	// Sleep 15 seconds before sending first heartbeat. This will give time to
    // properly finish TLS negotiation and then start sending heartbeats.
	sleepMills(15000);

	while( !packetWriter->done){
	//TODO  if (System.currentTimeMillis() - lastActive >= delay) ,there should be a mutex in writer.
	// Send heartbeat if no packet has been sent to the server for a given time
		if ( true ) {
			*(packetWriter->writer) << " ";
			packetWriter->writer->flush();
		}
	}
	sleepMills(delay);
}

PacketWriter::ListenerWrapper::ListenerWrapper(PacketListener * packetListener, PacketFilter * packetFilter) :
	packetListener(packetListener),
	packetFilter(packetFilter)
{
}

void PacketWriter::ListenerWrapper::notifyListener(Packet * packet) {
    if (packetFilter == NULL || packetFilter->accept( packet)) {
    	packetListener->processPacket( packet);
    }
}

PacketWriter::InterceptorWrapper::InterceptorWrapper(PacketInterceptor * packetInterceptor, PacketFilter * packetFilter) :
	packetInterceptor(packetInterceptor),
	packetFilter(packetFilter)
{
}

void PacketWriter::InterceptorWrapper::notifyListener(Packet * packet) {
    if (packetFilter == NULL || packetFilter->accept( packet)) {
    	packetInterceptor->interceptPacket( packet);
    }
}

void PacketWriter::openStream() {
    *writer << "<stream:stream";
    *writer << " to=\"" << connection->getServiceName() << "\"";
    *writer << " xmlns=\"jabber:client\"";
    *writer << " xmlns:stream=\"http://etherx.jabber.org/streams\"";
    *writer << " version=\"1.0\">";
    writer->flush();
	
}

